package com.ylkj.bus.app.report;


import com.ylkj.dao.RdsConfig;
import com.ylkj.dao.RdsUtils;
import com.ylkj.redis.RedisConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.*;
import java.text.MessageFormat;
import java.util.Arrays;
import java.util.List;

/**
 * @Author: wengzx
 * @Date: 2025/3/22
 * @Desc: 处理t_iotmo_app表数据(先删除后插入)
 */
@Slf4j
public class UnpackIotmoAppDeleteAndInsert extends ProcessFunction<List<IotmoAppDO>,List<IotmoAppDO>> {
    //基础删除SQL模板
    private static final String BASE_DELETE_SQL = "DELETE FROM {0} WHERE device_identifier = ? AND container_name = ?";

    private static final String INSERT_SQL =
            "INSERT INTO {0} (`id`, `device_identifier`, `app_name`, `app_version`, `container_name`, `service_status`, `enable_status`, create_time, update_time) " +
            "VALUES (?, ?, ?, ?, ?, ?, ?, now(), now()) " +
            "ON DUPLICATE KEY UPDATE " +
            "`device_identifier` = VALUES(`device_identifier`), " +
            "`app_name` = VALUES(`app_name`), " +
            "`app_version` = VALUES(`app_version`), " +
            "`container_name` = VALUES(`container_name`), " +
            "`service_status` = VALUES(`service_status`), " +
            "`enable_status` = VALUES(`enable_status`), " +
            "update_time = now()";
    // 数据库配置和表名
    private final String tableName;
    private final RdsConfig rdsConfig;
    private RdsUtils rdsUtils;

    // 格式化后的基础SQL语句
    private String baseSql;

    private String insertOnDuplicateSql;

    private static final int DEADLOCK_ERROR_CODE = 1213;

    //构造函数，用于初始化
    public UnpackIotmoAppDeleteAndInsert(String tableName , RdsConfig rdsConfig) {
        this.rdsConfig = rdsConfig;
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        rdsUtils = new RdsUtils(rdsConfig);

        baseSql = MessageFormat.format(BASE_DELETE_SQL, tableName);
        insertOnDuplicateSql = MessageFormat.format(INSERT_SQL, tableName);

    }

    @Override
    public void processElement(List<IotmoAppDO> iotmoAppDOS, ProcessFunction<List<IotmoAppDO>, List<IotmoAppDO>>.Context ctx, Collector<List<IotmoAppDO>> out) throws Exception {
        if (iotmoAppDOS == null || iotmoAppDOS.isEmpty()) {
            return;
        }
        IotmoAppDO iotmoAppDO = iotmoAppDOS.get(0);
        String deviceIdentifier = iotmoAppDO.getDeviceIdentifier();
        String containerName = iotmoAppDO.getContainerName();
        int retryCount = 0;
        try {
            retryCount = 0;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

        while (true) {
            try (Connection connection = rdsUtils.getConnection()) {
                connection.setAutoCommit(false); // 开启事务
                try {
                    boolean existsAndDeleted = deleteIfExists(connection, deviceIdentifier, containerName);
                    if (existsAndDeleted) {
                        log.info("已删除旧记录，deviceIdentifier: {}, containerName: {}", deviceIdentifier, containerName);
                    }
                    int insertedRows = insertBatchData(connection, iotmoAppDOS);
                    log.info("成功插入新记录，deviceIdentifier: {}, containerName: {}", deviceIdentifier, containerName);
                    connection.commit(); // 提交事务
                    break; // 成功退出循环
                } catch (SQLException e) {
                    if (e.getErrorCode() == DEADLOCK_ERROR_CODE && retryCount < 3) {
                        retryCount++;
                        log.warn("死锁重试 {} 次: {}", retryCount, e.getMessage());
                        continue; // 重试事务
                    } else {
                        connection.rollback();
                        log.error("数据库操作失败，已回滚事务。错误信息: {}", e.getMessage(), e);
                        throw e;
                    }
                }
            }

        }
    }

    /**
     *  删除指定 deviceIdentifier 和  containerName 的数据
     * @param connection
     * @param deviceIdentifier
     * @param containerName
     * @return
     * @throws SQLException
     */
    private boolean deleteIfExists(Connection connection, String deviceIdentifier, String containerName) throws SQLException {
        try (PreparedStatement stmt = connection.prepareStatement(baseSql)) {
            stmt.setString(1, deviceIdentifier);
            stmt.setString(2, containerName);
            int affectedRows = stmt.executeUpdate();
            return affectedRows > 0; // 如果删除了数据，返回 true
        }
    }

    /**
     *  批量插入或更新数据
     * @param connection
     * @param iotmoAppDOS
     * @return
     * @throws SQLException
     */
    private int insertBatchData(Connection connection, List<IotmoAppDO> iotmoAppDOS) throws SQLException {
        String sql = MessageFormat.format(INSERT_SQL, tableName);
        try (PreparedStatement stmt = connection.prepareStatement(sql)) {
            int batchSize = 0;

            for (IotmoAppDO item : iotmoAppDOS) {
                try {
                    stmt.setString(1, item.getId());
                    stmt.setString(2, item.getDeviceIdentifier());
                    stmt.setString(3, item.getAppName());
                    stmt.setString(4, item.getAppVersion());
                    stmt.setString(5, item.getContainerName());
                    stmt.setString(6, item.getServiceStatus());
                    stmt.setString(7, item.getEnableStatus());
                    stmt.addBatch();
                    if (++batchSize % 5 == 0) { // 每5条提交一次
                        stmt.executeBatch();
                    }
                }catch (SQLException e) {
                    log.error("插入或更新记录时出错，跳过。ID：{}，错误：{}", item.getId(), e.getMessage(), e);
                }
            }
            stmt.executeBatch(); // 插入剩余数据
            return batchSize;
        }
    }
    @Override
    public void close() throws Exception {
        rdsUtils.close();
        super.close();
    }
}
